package com.leilei.join.window;

import com.leilei.join.common.VehicleDetail;
import com.leilei.join.common.VehicleInfo;
import org.apache.flink.api.common.eventtime.*;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 21:52
 * @desc 自定义车辆详情水位线设置 默认以系统时间为时间时间
 */
public class VehicleDetailWatermark implements WatermarkStrategy<VehicleDetail> {
    @Override
    public TimestampAssigner<VehicleDetail> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return (element, recordTimestamp) -> System.currentTimeMillis();
    }

    @Override
    public WatermarkGenerator<VehicleDetail> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new WatermarkGenerator<VehicleDetail>() {
            @Override
            public void onEvent(VehicleDetail event, long eventTimestamp, WatermarkOutput output) {
                output.emitWatermark(new Watermark(System.currentTimeMillis()));
            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                output.emitWatermark(new Watermark(System.currentTimeMillis()));
            }
        };
    }
}
